package org.couchbase;

import java.io.IOException;
import java.net.SocketAddress;
import java.net.URI;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.TimeUnit;

import net.spy.memcached.CASValue;
import net.spy.memcached.internal.OperationFuture;

import org.acquaintance.I_CRUD;
import org.codehaus.jettison.json.JSONException;
import org.codehaus.jettison.json.JSONObject;

import com.couchbase.client.CouchbaseClient;
import com.couchbase.client.CouchbaseConnectionFactoryBuilder;
import com.couchbase.client.protocol.views.DesignDocument;
import com.couchbase.client.protocol.views.Query;
import com.couchbase.client.protocol.views.Stale;
import com.couchbase.client.protocol.views.View;
import com.couchbase.client.protocol.views.ViewDesign;
import com.couchbase.client.protocol.views.ViewResponse;
import com.couchbase.client.protocol.views.ViewRow;

public class CouchbaseFacade implements I_CRUD
{
	private CouchbaseClient client = null;
	private final int expireTime = 3600;
	private String host = "ec2-54-246-45-82.eu-west-1.compute.amazonaws.com";
//	private String host = "localhost";

	private String nameOfdesignDoc_get = "benchmark_get";
	private String nameOfdesignDoc_ids = "benchmark_ids";
	private String nameOfView_get = "get_all";
	private String nameOfView_ids = "get_ids";
	
	/**
	 * Open and close the connection to Couchbase
	 */
	private void openAccessToCouchbase()
	{
	    List<URI> uris = new LinkedList<URI>();
	    uris.add(URI.create("http://" + host + ":8091/pools"));
	    
		try
		{
			CouchbaseConnectionFactoryBuilder cfb = new CouchbaseConnectionFactoryBuilder();
			cfb.setOpTimeout(500000);
			client = new CouchbaseClient(cfb.buildCouchbaseConnection(uris, "default", ""));
		}
		catch (IOException e)
		{
			System.err.println("IOException connecting to Couchbase: " + e.getMessage());
			System.exit(1);
		}
	}
	
	private void closeAccessToCouchbase()
	{
		client.shutdown(3, TimeUnit.SECONDS);
	}
	
	/**
	 * CRUD operations
	 */
	@Override
	public void create(int repetitions, String attributeName, String attributeValue)
	{
		JSONObject jsonDoc = new JSONObject();
		try
		{
			jsonDoc.put(attributeName, attributeValue);
			String toPut = jsonDoc.toString();
			
			openAccessToCouchbase();

			int baseline_ep_total_persisted = getValueOfStat(1);
			long start = 0;
			for (Integer i = 0; i < repetitions; i++)
			{
				OperationFuture<Boolean> success = client.set(i.toString(), expireTime, toPut);
//				client.set(i.toString(), expireTime, toPut);
				if (i==0)
					start = System.currentTimeMillis();
				while (!success.isDone())
					;
//				try
//				{
//					Thread.sleep(1);
//				}
//				catch (InterruptedException e)
//				{ e.printStackTrace(); }
			}
			
			int goal_ep_total_persisted = repetitions + baseline_ep_total_persisted;
			do
			{
				int current_ep_total_persisted = getValueOfStat(1);
				if (current_ep_total_persisted == goal_ep_total_persisted)
				{
					long end = System.currentTimeMillis();
					System.out.println("Time: " + Long.toString(end-start) + " milli seconds");
					break;
				}
				
				try
				{
					Thread.sleep(500);
				}
				catch (InterruptedException e)
				{ e.printStackTrace(); }
			} while (true);
			
			closeAccessToCouchbase();
		}
		catch (JSONException e)
		{
			e.printStackTrace();
		}
	}
	
	@Override
	public void read()
	{
		createViews();
		try
		{
			openAccessToCouchbase();
			
			View view = client.getView(nameOfdesignDoc_get, nameOfView_get);
			Query query = new Query();
			query.setStale(Stale.FALSE);
			query.setIncludeDocs(true);
			
			long startTimeRead = System.currentTimeMillis();
			ViewResponse response = client.query(view, query);
			long endTimeRead = System.currentTimeMillis();
			
			System.out.println("Time: " + (endTimeRead - startTimeRead)  + " milli seconds");
			System.out.println("Records found: " + response.size());
		}
		finally
		{
			closeAccessToCouchbase();
		}
	}

	@Override
	public void update(String attributeName, String attributeValue)
	{
		createViews();
		try
		{
			openAccessToCouchbase();
			
			int baseline_vb_active_ops_update = getValueOfStat(3);
			
			View view = client.getView(nameOfdesignDoc_get, nameOfView_get);
			Query query = new Query();
			query.setStale(Stale.FALSE);
			query.setIncludeDocs(true);
			
			long start = System.currentTimeMillis();
			ViewResponse response = client.query(view, query);
			
			for(ViewRow row : response)
			{
				JSONObject toUpdate = new JSONObject((String)row.getDocument());
				toUpdate.put(attributeName, attributeValue);
				OperationFuture<Boolean>success = client.replace(row.getId(), expireTime, toUpdate.toString());
				while (!success.isDone())
					;
//				client.replace(row.getId(), expireTime, toUpdate.toString());
//				try
//				{
//					Thread.sleep(1);
//				} catch (InterruptedException e)
//				{ e.printStackTrace(); }
			}
			
			int goal_vb_active_ops_update = response.size() + baseline_vb_active_ops_update;
			
			do
			{
				int current_vb_active_ops_update = getValueOfStat(3);
				
				if (current_vb_active_ops_update == goal_vb_active_ops_update)
				{
					long end = System.currentTimeMillis();
					System.out.println("Time: " + Long.toString(end-start) + " milli seconds");
					break;
				}
				try
				{
					Thread.sleep(400);
				}
				catch (InterruptedException e)
				{ e.printStackTrace(); }
			} while (true);
			closeAccessToCouchbase();
		}
		catch (JSONException e)
		{
			e.printStackTrace();
		}
	}

	@Override
	public void delete()
	{
		createViews();
		
		openAccessToCouchbase();
		/**
		 * Get all current keys in the bucket
		 * Put into an ArrayList, to be used when doing the actual delete
		 */
		View view = client.getView(nameOfdesignDoc_ids, nameOfView_ids);
		Query query = new Query();
		query.setStale(Stale.FALSE);
		ViewResponse response = client.query(view, query);
		
		ArrayList<String> keysToDelete = new ArrayList<String>();
		
		for(ViewRow row : response)
		{
			keysToDelete.add(row.getId());
		}
		
		boolean first = true;
		int baseline_ep_total_del_items = getValueOfStat(2);
//		System.out.println("base: " + baseline_ep_total_del_items);
		long start = 0;
		for (String key : keysToDelete)
		{			
			client.delete(key);
			
			if (first)
			{
				start = System.currentTimeMillis();
				first = false;
			}
		}
		int goal_ep_total_del_items = keysToDelete.size() + baseline_ep_total_del_items;
		
		do
		{
			int current_ep_total_del_items = getValueOfStat(2);
			if (current_ep_total_del_items == goal_ep_total_del_items)
			{
				long end = System.currentTimeMillis();
				System.out.println("Time: " + Long.toString(end-start) + " milli seconds");
				break;
			}
			try
			{
				Thread.sleep(400);
			}
			catch (InterruptedException e)
			{ e.printStackTrace(); }
		} while (true);
		closeAccessToCouchbase();
	}

	/**
	 * Private methods
	 */
	@SuppressWarnings({ "unchecked", "rawtypes" })
	private void createViews()
	{
		openAccessToCouchbase();
		
		try
		{
			client.getView(nameOfdesignDoc_get, nameOfView_get);
		}
		catch (com.couchbase.client.protocol.views.InvalidViewException e)
		{
			System.out.println("View " + nameOfView_get + " does not exist. Creating it now");
			
			DesignDocument designDoc = new DesignDocument(nameOfdesignDoc_get);
			
			String viewName = nameOfView_get;
			String mapFunction =
					"function (doc, meta) {\n" +
					"  emit(meta.id, doc); \n" +
					"}";
			
			ViewDesign viewDesign = new ViewDesign(viewName,mapFunction);
			
			designDoc.getViews().add(viewDesign);
			client.createDesignDoc(designDoc);
		}
		
		try
		{
			client.getView(nameOfdesignDoc_ids, nameOfView_ids);
		}
		catch (com.couchbase.client.protocol.views.InvalidViewException e)
		{
			System.out.println("View " + nameOfView_ids + " does not exist. Creating it now");
			
			DesignDocument designDoc = new DesignDocument(nameOfdesignDoc_ids);
			
			String viewName = nameOfView_ids;
			String mapFunction =
					"function (doc, meta) {\n" +
					"  emit(meta.id); \n" +
					"}";
			
			ViewDesign viewDesign = new ViewDesign(viewName,mapFunction);
			
			designDoc.getViews().add(viewDesign);
			client.createDesignDoc(designDoc);
		}
		
		closeAccessToCouchbase();
	}
	
	private int getValueOfStat(int number)
	{
		String stat_ep_total_persisted = "ep_total_persisted",
			   stat_ep_total_del_items = "ep_total_del_items",
			   stat_vb_active_ops_update = "vb_active_ops_update";
		int ep_total_persisted = 0,
			ep_total_del_items = 0,
			vb_active_ops_update = 0;
		
		Map<SocketAddress, Map<String, String>> stats = client.getStats();

		for (Entry<SocketAddress, Map<String, String>> entry : stats.entrySet())
		{
			Map<String, String> currentEntry = stats.get(entry.getKey());
			try { ep_total_persisted = Integer.parseInt(currentEntry.get(stat_ep_total_persisted)); }
			catch (NumberFormatException nEx) { ep_total_persisted = 0; }
			try { ep_total_del_items = Integer.parseInt(currentEntry.get(stat_ep_total_del_items)); }
			catch (NumberFormatException nEx) { ep_total_del_items = 0; }
			try { vb_active_ops_update += Integer.parseInt(currentEntry.get(stat_vb_active_ops_update)); }
			catch (NumberFormatException nEx) { vb_active_ops_update = 0; }
		}
		if (number == 1)
			return ep_total_persisted;
		if (number == 2)
			return ep_total_del_items;
		else
			return vb_active_ops_update;
	}
}
